package com.chinaway.mobile.container;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.alibaba.fastjson.JSONObject;
import com.chinaway.mobile.dao.BalanceCounterDao;
import com.chinaway.mobile.http.util.ApiUpdateResult;
import com.chinaway.mobile.http.util.ConstantUtil;
import com.chinaway.mobile.http.util.ServiceUtils;
import com.chinaway.mobile.model.BalanceCounter;
import com.chinaway.mobile.model.MobileBean;
import com.chinaway.mobile.model.ReqMobileBean;
import com.chinaway.mobile.model.ResultMobileBean;
import com.chinaway.mobile.thread.ComplexThreadPool;
import com.chinaway.mobile.thread.MobileTask;
import com.chinaway.proxy.ProxyService;

public class HttpDataContainer {
	private BalanceCounterDao balanceCounterDao;
	private final static Log log = LogFactory.getLog(HttpDataContainer.class);
	private final static long limitTime = 1000 * 60 * 5;
	protected  ProxyService proxyService;
	private LinkedBlockingQueue<MobileBean> requestData = new LinkedBlockingQueue<MobileBean>(
			100000);
	private LinkedBlockingQueue<Future<ResultMobileBean>> tasksResult = new LinkedBlockingQueue<Future<ResultMobileBean>>();
	private ConcurrentHashMap<String, List<ResultMobileBean>> queueMap = new ConcurrentHashMap<String, List<ResultMobileBean>>();
	private ScheduledThreadPoolExecutor exec_tasksResult = new ScheduledThreadPoolExecutor(
			2);// 定时任务线程池,监控tasksResult
////	private ScheduledThreadPoolExecutor exec_map = new ScheduledThreadPoolExecutor(
////			2);// 定时任务线程池,监控tasksResult
	
	
	public void setProxyService(ProxyService proxyService) {
		this.proxyService = proxyService;
		log.info("HttpDataContainer proxyService 初始化成功："+proxyService);
	}
	

	// 启动定时任务
	public void init() {
		// 当前一个定时任务还么执行完，下一个定时任务不会执行
		exec_tasksResult.scheduleAtFixedRate(new Runnable() {
			@Override
			public void run() {
				int success = 0;
				boolean flag = false;
				long start = System.currentTimeMillis();
				while (tasksResult.size() > 0) {
					Future<ResultMobileBean> future = null;
					flag = true;
					try {
						ResultMobileBean rmb = null;
						future = tasksResult.peek();
						if (future != null && future.isDone()) {
							rmb = future.get();
							if (StringUtils.isNotBlank(rmb.getBalance())) {
								success++;
							}
							tasksResult.remove(future);
							put2QueueMap(rmb);
						} else if (future.isCancelled()) {// 如果任务自动取消，则移除队列
							tasksResult.remove(future);
							log.info("查询手机余额任务自动取消，移除队列！");
						}
						
						// 判断使用是否超时
						long interval = System.currentTimeMillis() - start;
						if (interval >= limitTime) {
							int count = tasksResult.size();
							tasksResult.clear();
							log.info("tasksResult处理超时，tasksResult移除：" + count
									+ " tasksResult.size=" + tasksResult.size());
							break;
						}

					} catch (InterruptedException e) {
						log.info("线程中断！" + e.toString() + ",Method="
								+ e.getStackTrace()[0].getMethodName()
								+ ",Line="
								+ e.getStackTrace()[0].getLineNumber());
						tasksResult.remove(future);
					} catch (ExecutionException e) {
						log.info("线程异常！" + e.toString() + ",Method="
								+ e.getStackTrace()[0].getMethodName()
								+ ",Line="
								+ e.getStackTrace()[0].getLineNumber());
						tasksResult.remove(future);
					} catch (Exception e) {
						log.info("线程异常！" + e.toString() + ",Method="
								+ e.getStackTrace()[0].getMethodName()
								+ ",Line="
								+ e.getStackTrace()[0].getLineNumber());
						tasksResult.remove(future);
					}
				}
				if (flag){
					log.info("手机余额任务结束！成功查询：" + success + "条数据");
					balanceCounterDao.save(new BalanceCounter(new Date(), "成功查询", success, "1"));
				}
				
				//发送给php
				send();

				// 传数据给线程池
				addTasks();
				log.info("本次定时任务结束！");
				
			}
		}, 0, ConstantUtil.EXEC_TASKRESULT_TIME, TimeUnit.SECONDS);
		
		// 定时发送
//		exec_map.scheduleAtFixedRate(new Runnable() {
//			@Override
//			public void run() {
//				// 遍历map，发送更新请求
//				if (queueMap != null && queueMap.size() > 0) {
//					for (Entry<String, List<ResultMobileBean>> entry : queueMap
//							.entrySet()) {
//						List<ResultMobileBean> current = entry.getValue();
//						String key = entry.getKey();
//						updatePHPDate(key, current);
//						// 删除
//						current.clear();
//						queueMap.remove(key);
//					}
//				}
//			}
//		}, 0, ConstantUtil.EXEC_MAP_TIME, TimeUnit.SECONDS);
	}
	
	public void send(){
		log.info("刷新手机余额queueMap！");
		// 遍历map，发送更新请求
		if (queueMap != null && queueMap.size() > 0) {
			for (Entry<String, List<ResultMobileBean>> entry : queueMap
					.entrySet()) {
				List<ResultMobileBean> current = entry.getValue();
				String key = entry.getKey();
				updatePHPDate(key, current);
				// 删除
				current.clear();
				queueMap.remove(key);
			}
		}
		log.info("刷新手机余额queueMap结束！");
	}


	// 存请求来的数据
	public void put2MobileCard(List<MobileBean> mbs) {
		// 存入队列

		for (MobileBean mobileBean : mbs) {
			if (!requestData.add(mobileBean)) {
				log.error(mobileBean.getMobile() + "加入队列失败," + "Size:"
						+ requestData.size());
			}
		}
		log.info("[put2MobileCard]成功接收MobileBean.size=" + mbs.size()
				+ " requestData.size=" + requestData.size()
				+ ",tasksResult.size:" + tasksResult.size() + ",queueMap.zise:"
				+ queueMap.size());
		balanceCounterDao.save(new BalanceCounter(new Date(), "成功接收", mbs.size(), "2"));
	}

	// 回调PHP，更新数据,path=url-key-method
	public void updatePHPDate(String path,
			List<ResultMobileBean> resultdata) {
		log.info("PHP更新size：" + resultdata.size());
		String[] ukm = path.split("-");
		Map<String, Object> params = new HashMap<String, Object>();
		ApiUpdateResult result = new ApiUpdateResult();
		List<ReqMobileBean> rs = new ArrayList<ReqMobileBean>();
		for (ResultMobileBean resultmb : resultdata) {
			if(resultmb.getMessage()!=null&&resultmb.getMessage().toLowerCase().contains("html")){
				rs.add(new ReqMobileBean(resultmb.getMobile(), resultmb
						.getBalance(), resultmb.getSuccess(), "网络错误！"));
			}else{
				rs.add(new ReqMobileBean(resultmb.getMobile(), resultmb
						.getBalance(), resultmb.getSuccess(), resultmb.getMessage()));
			}
		}
		result.setData(rs);
		String resultend = JSONObject.toJSONString(result);
		log.info("本次回调请求数据："+resultend);
		params.put("result", resultend);
		ServiceUtils.call(ukm[0], ukm[1], ukm[2], params);
		balanceCounterDao.save(new BalanceCounter(new Date(), "成功发送", rs.size(), "3"));
		log.info("PHP更新结束");
	}

	// 获取余额后分组存入queueMap
	public synchronized void put2QueueMap(ResultMobileBean rmb) {
		if (queueMap == null) {
			queueMap = new ConcurrentHashMap<String, List<ResultMobileBean>>();
		}
		String rmb_key = rmb.getAppurl() + "-" + rmb.getAppkey() + "-"
				+ rmb.getMethod();
		boolean flag = false;// false表示没有key对应，需要创建list
		// 添加到map
		for (Entry<String, List<ResultMobileBean>> entry : queueMap.entrySet()) {
			List<ResultMobileBean> current = entry.getValue();
			String key = entry.getKey();
			// 存入queueMap
			if (key.equals(rmb_key)) {
				current.add(rmb);
				flag = true;// 已加入
				break;
			}
		}
		if (!flag) {
			List<ResultMobileBean> newlist = new ArrayList<ResultMobileBean>();
			newlist.add(rmb);
			queueMap.put(rmb_key, newlist);
		}
	}

	// 添加任务到线程池
	private void addTasks() {
		log.info("执行增加任务");
		if (requestData.size() > 0) {
			int max_count = requestData.size() > 100 ? 100 : requestData.size();
			Future<ResultMobileBean> future =null;
			MobileBean mobileBean =null;
			for (int i = 0; i < max_count; i++) {
				try {
				    mobileBean = requestData.poll();
					if (mobileBean != null) {
						future = ComplexThreadPool.TASK_COMMISSION_POOL
								.submit((new MobileTask(mobileBean.getMobile(),
										mobileBean.getPassword(), mobileBean
												.getCity(), mobileBean
												.getAppurl(), mobileBean
												.getAppkey(), mobileBean
												.getMethod(),proxyService)));
						tasksResult.add(future);
					}

				} catch (Exception e) {
					tasksResult.remove(future);
					log.info("增加任务异常,mobile="+mobileBean.getMobile()+",reason" + e.toString() + ",Method="
							+ e.getStackTrace()[0].getMethodName()
							+ ",Line="
							+ e.getStackTrace()[0].getLineNumber());
				}
			}
			log.info("增加任务 requestData size:" + max_count);
		}
		log.info("执行增加任务结束");
	}

	public BalanceCounterDao getBalanceCounterDao() {
		return balanceCounterDao;
	}

	public void setBalanceCounterDao(BalanceCounterDao balanceCounterDao) {
		this.balanceCounterDao = balanceCounterDao;
	}

	public String Compute(Date from,Date to,int type){
		List<BalanceCounter> list=balanceCounterDao.select(from, to, type);
		int sum=0;
		for (BalanceCounter bc : list) {
			sum+=bc.getCount();
		}
		/*
		 * 成功查询 = 1 
		 * 成功接收 = 2
		 * 成功发送 = 3
		 */
		switch (type) {
		case 1:
			return "有 "+list.size()+" 条记录，一共成功查询: "+sum;
		case 2:
			return "有 "+list.size()+" 条记录，一共成功接收: "+sum;
		case 3:
			return "有 "+list.size()+" 条记录，一共成功发送: "+sum;

		default:
			return "type不正确";
		}
	}
	
	public void destroy() {
		exec_tasksResult.shutdownNow();
		log.info("exec_tasksResult成功关闭定时任务！");
	}
	public void start_exec_tasksResult(){
		exec_tasksResult.shutdownNow();
		init();
		log.info("重新开启手机余额定时任务！");
	}
}
